package com.bw.yk06;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class BrodcastProcess extends BroadcastProcessFunction<String,String,String> {
    MapStateDescriptor<String, String> mapStateDescriptor =null;

    private boolean t;

    public BrodcastProcess(MapStateDescriptor<String, String> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    Connection conn =null;
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181", "", "");
        conn.setAutoCommit(true);
        t=true;
    }
    @Override
    public void close() throws Exception {
        conn.close();
    }




    @Override
    public void processElement(String value, BroadcastProcessFunction<String, String, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {
        if(t)
        {
            Thread.sleep(5000L);
            t=false;
        }

        //广播流
        ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //主流
        JSONObject jsonObject = JSON.parseObject(value);
        //根据source_table 得到sinktable
        String source_table = jsonObject.getString("table");
        String sinkTable = broadcastState.get(source_table);
        //存入hbase
        if(sinkTable!=null){
            PreparedStatement ps = conn.prepareStatement("upsert into " + sinkTable + " values (?,?)");
            ps.setObject(1,jsonObject.getJSONObject("data").getString("id"));
            ps.setObject(2,jsonObject.getJSONObject("data").toJSONString());
            ps.executeUpdate();
            ps.close();
        }

    }

    @Override
    public void processBroadcastElement(String value, BroadcastProcessFunction<String, String, String>.Context ctx, Collector<String> out) throws Exception {
        //配置流
        JSONObject data = JSON.parseObject(value).getJSONObject("data");
        //得到source_table 和 sink_table
        String source_table = data.getString("source_table");
        String sink_table = data.getString("sink_table");
        //存入广播流
        BroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        broadcastState.put(source_table,sink_table);
        //生成表
        String sql="create table if not exists " + sink_table +" (id varchar primary key , info.line varchar) column_encoded_bytes=0";
        PreparedStatement ps = conn.prepareStatement(sql);
        //System.out.println("创建成功");
        ps.executeUpdate();

    }
}
